package com.hw.nio.netty.rpc;

import com.hw.nio.netty.rpc.test.UserService;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.concurrent.DefaultPromise;
import io.netty.util.concurrent.Promise;
import org.objectweb.asm.Type;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicLong;

/**
 * rpcClient
 */
public class RpcClient {

    private Channel channel;
    private Map<Long, Promise<Response>> results = new HashMap<>();
    // 生成ID
    static AtomicLong atomicLong = new AtomicLong(100);

    public void init(String address, int port) throws InterruptedException {
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(new NioEventLoopGroup())
                .channel(NioSocketChannel.class);
        bootstrap.handler(new ChannelInitializer<NioSocketChannel>() {
            @Override
            protected void initChannel(NioSocketChannel ch) throws Exception {
                ch.pipeline().addLast("codec", new RpcCodec());
                ch.pipeline().addLast("resultFill", new ResultFill());
            }
        });

        ChannelFuture connect = bootstrap.connect(address, port);
        channel = connect.sync().channel();
        System.out.println("连接成功");
        // 每两秒发送一次心跳
//        channel.eventLoop().scheduleWithFixedDelay(() -> {
//            long nextId = getNextId();
//            Transfer transfer = new Transfer(nextId);
//            transfer.heartbeat = true;
//            channel.writeAndFlush(transfer);
//        }, 2000, 2000, TimeUnit.MILLISECONDS);
    }


    public static long getNextId() {
        return atomicLong.get();
    }

    public <T> T getRemoteService(Class<T> tClass) {

        assert tClass.isInterface();
        Object proxyObj = Proxy.newProxyInstance(getClass().getClassLoader(), new Class[]{tClass}, new InvocationHandler() {
            @Override
            public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
                if (Object.class.equals(method.getDeclaringClass())) {
                    return method.invoke(args, proxy);
                }
                String methodDesc = method.getName() + Type.getMethodDescriptor(method);
                Response response = invokeRemoteMethod(tClass, methodDesc, args);
                if (response.getError() != null) {
                    throw new RuntimeException("远程服务调用异常", response.getError());
                }
                return response.getResult();
            }
        });
        return (T) proxyObj;
    }

    private <T> Response invokeRemoteMethod(Class<T> tClass, String methodDesc, Object[] args)
            throws InterruptedException, ExecutionException, TimeoutException {
        Request request = new Request(methodDesc, tClass.getName());
        request.setArgs(args);
        Transfer transfer = new Transfer(getNextId());
        transfer.serializableId = Transfer.SERIALIZABLE_JAVA;
        transfer.request = true;
        transfer.target = request;
        DefaultPromise<Response> promise = new DefaultPromise<>(channel.eventLoop());
        channel.writeAndFlush(transfer).addListener(future -> {
            if (future.cause() != null) {
                promise.setFailure(future.cause());
            } else {
                results.put(transfer.id, promise);
            }
        });

        return promise.get(10, TimeUnit.SECONDS);
    }

    private class ResultFill extends SimpleChannelInboundHandler<Transfer> {
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, Transfer msg) throws Exception {
            if (msg.heartbeat) {
                System.out.println(String.format("服务端心跳返回：%s",
                        ctx.channel().remoteAddress()));
            } else {
                Promise<Response> promise = results.remove(msg.id);
                if (promise != null) {
                    promise.setSuccess((Response) msg.target);
                }

            }

        }
    }
}
